package com.gitee.jnxj.nraft.election;

import com.gitee.jnxj.nraft.common.core.*;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.example.nrpc.client.proxy.BeanFactory;
import org.example.nrpc.client.proxy.ChannelInActiveException;
import org.example.nrpc.client.proxy.ChannelNotFoundException;
import org.example.nrpc.common.model.RpcAddress;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;

/**
 * 选举
 *
 * @author 江南小俊
 * @since 2021/7/12
 **/
@RequiredArgsConstructor
@Slf4j
public class Election {
    @Getter
    private static Election instance;
    @Getter
    private Node node;
    private ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
    private ScheduledFuture<?> scheduledFuture;
    private ScheduledFuture<?> heartbeatFuture;


    public Election(Node node) {
        this.node = node;
        Election.instance = this;
    }

    /**
     * 初始节点选举超时时间(3-5s)<br/>
     * 添加选举任务
     *
     * @return void
     * @author Jim
     * @since 2021/7/12 上午11:45
     **/

    @PostConstruct
    public void init() {
        log.debug("init:{}", node);
        addElectionTask();
    }

    /**
     * 初始节点选举超时时间(3-5s)<br/>
     * 添加选举任务
     *
     * @return void
     * @author Jim
     * @since 2021/7/12 上午11:47
     **/

    private void addElectionTask() {
        log.debug("添加选举任务");
        this.node.setElectionAlarm((int) (Math.random() * 3000 + 8000));
        scheduledFuture = service.schedule(() -> {
            votedForSelf();
            requestToVote();
            addElectionTask();
        }, this.node.getElectionAlarm(), TimeUnit.MILLISECONDS);
    }

    /**
     * 重置选举任务
     *
     * @return void
     * @author Jim
     * @since 2021/7/12 上午11:47
     **/

    public void resetElectionTask() {
        //取消之前的选举任务
        scheduledFuture.cancel(true);
        if (heartbeatFuture != null) heartbeatFuture.cancel(true);
        if (scheduledFuture.isCancelled()) {
            addElectionTask();
        } else {
            log.error("取消选举任务失败");
        }
    }

    /**
     * 请求其它节点进行投票
     *
     * @return void
     * @author Jim
     * @since 2021/7/12 上午11:39
     **/

    private void requestToVote() {
        for (Peer peer : node.getPeers()) {
            //rpc
            log.info("请求Node-{}-投票", peer.getId());
            VoteReq voteParam = VoteReq.builder()
                    .term(node.getCurrentTerm())
                    .candidateId(node.getId())
                    .commitIndex(node.getCommitIndex())
                    .build();

            RaftService electionService = null;
            try {
                electionService = getElectionService(peer);
            } catch (ChannelInActiveException | ChannelNotFoundException e) {
//                log.error("", e);
                continue;
            }
            electionService.requestVote(voteParam).addListener((future) -> {
                if (future.isDone() && node.getType().equals(Node.NodeType.CANDIDATE)) {
                    try {
                        Boolean agree = (Boolean) future.get();
                        log.info("请求投票结果：{}", agree);
                        if (agree) {
                            peer.setVoteGranted(true);
                            checkVoteResult();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

    }

    private RaftService getElectionService(Peer peer) throws ChannelNotFoundException, ChannelInActiveException {
        RpcAddress server = new RpcAddress();
        server.setHost(peer.getPeerAddress().getAddress());
        server.setPort(peer.getPeerAddress().getPort());
        RaftService electionService;
        electionService = BeanFactory.getBean(server, RaftService.class);
        return electionService;
    }

    /**
     * 检查投票情况<br/>
     *
     * @return void
     * @author Jim
     * @since 2021/7/13 上午10:54
     **/

    private void checkVoteResult() {
        if (node.getType() == Node.NodeType.CANDIDATE) {
            synchronized (this) {
                if (node.getType() == Node.NodeType.CANDIDATE) {
                    long agreeNum = node.getPeers().stream().filter(Peer::isVoteGranted).count();
                    if (node.getId() == node.getVotedFor()) agreeNum++;
                    if (agreeNum * 2 > node.getPeers().size() + 1) {
                        //过半同意
                        log.info("过半同意，Node-{}-成为Leader", node.getId());
                        node.setType(Node.NodeType.LEADER);
                        scheduledFuture.cancel(true);
                        heartbeat();
                    }
                }
            }
        }
    }

    /**
     * 开始心跳
     *
     * @return void
     * @author Jim
     * @since 2021/7/13 上午11:44
     **/

    private void heartbeat() {
        heartbeatFuture = service.scheduleAtFixedRate(() -> {
            for (Peer peer : node.getPeers()) {
                if (node.getType() != Node.NodeType.LEADER) continue;
                try {
                    log.info("开始发送心跳:to-{}", peer.getId());
                    RaftService electionService = getElectionService(peer);
                    electionService.heartbeat(HeartbeatReq.builder()
                            .index(node.getId())
                            .term(node.getCurrentTerm()).build()).addListener((heartbeatFuture -> {
                        if (heartbeatFuture.isDone()) {
                            try {
                                HeartbeatResp resp = (HeartbeatResp) heartbeatFuture.get();
                                if (!resp.isAgree()) {
                                    //重新开始新的选举
                                    node.setCurrentTerm(resp.getTerm());
                                    node.setType(Node.NodeType.FOLLOWER);
                                    resetElectionTask();
                                }
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            } catch (ExecutionException e) {
                                e.printStackTrace();
                            }
                        }
                    }));
                } catch (ChannelNotFoundException | ChannelInActiveException e) {
//                    log.error("", e);
                }
            }
        }, 0, 3, TimeUnit.SECONDS);
    }

    /**
     * 给当前节点投票<br/>
     * 节点类型从FOLLOWER->CANDIDATE<br/>
     * 任期+1<br/>
     * 投票赞成：当前节点
     *
     * @return void
     * @author Jim
     * @since 2021/7/12 上午11:39
     **/

    private void votedForSelf() {
        node.setType(Node.NodeType.CANDIDATE);
        node.setCurrentTerm(node.getCurrentTerm() + 1);
        node.setVotedFor(node.getId());
        log.info("开启新一轮选举:{}", node);
    }
}
